package com.arnold.demo.journal.main

import java.util.Properties

import com.alibaba.fastjson.{JSON, JSONObject}
import com.arnold.demo.journal.common.JedisUtil
import com.arnold.demo.journal.config.Config
import org.apache.kafka.clients.producer.{KafkaProducer, ProducerRecord}

/**
  * Created by arnold.zhu on 2017/7/20.
  */
object Redis2Kafka {

  def main(args: Array[String]): Unit = {
    val jedis = JedisUtil.getJedis
    val len = jedis.llen("redis_pos_journals")

    val props = new Properties()
    props.put("bootstrap.servers", Config.KAFKA_BROKERS)
    props.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    props.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer")
    val producer = new KafkaProducer[String, String](props)
    val errJsonObj = JSON.parseObject("{\"errorCode\":\"D9\",\"invoiceDate\":{\"date\":29,\"day\":2,\"hours\":17,\"minutes\":22,\"month\":10,\"seconds\":11,\"time\":1480411331000,\"timezoneOffset\":-480,\"year\":116},\"invoiceId\":\"\",\"invoiceType\":1,\"invoiceValue\":0,\"status\":0,\"storeNumber\":\"S26423\"}")

    var i = 0
    var count = 0 // 测试用计数变量
    while (i < 2000) {
      val str = jedis.lindex("redis_pos_journals", i)

      val jsonObj: JSONObject = JSON.parseObject(str)

      val message: String = jsonObj.getString("message")

      if (message.length > 0) {

        producer.send(new ProducerRecord[String, String](Config.KAFKA_JOURNAL_TOPIC, str))
        producer.send(new ProducerRecord[String, String](Config.KAFKA_ERROR_TOPIC, errJsonObj.toString))

        println("Message sent: " + str)
        println("Error Message sent: " + jsonObj.toString)

        Thread.sleep(500)
        count = count + 1
      }
      i = i + 1
      if (i == 2000) {
        i = 0
      }
    }
    println(count)
  }


}
